package mr.dividends;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import mr.dividends.model.DividendsKeyWritable;
import mr.dividends.model.DividendsValueWritable;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import au.com.bytecode.opencsv.CSVReader;

public class DividendsReducer extends Reducer<DividendsKeyWritable, DividendsValueWritable, Text, FloatWritable> {

	private static Logger LOG = LoggerFactory.getLogger(DividendsReducer.class);
	private final Text reusableText = new Text();
	private final FloatWritable reuableFloat = new FloatWritable();

	public final static String SYMBOL_DESCRIPTION_FILE = "symbol_description.csv";
	private final Map<String, String> stockNameLoopup = new HashMap<String, String>();

	enum Tokens {
		Enriched, NotEnriched
	}
	
	@Override
	protected void setup(Context context) throws InterruptedException {
		
		FileReader reader = null;
		CSVReader csvReader = null;
		try {
			reader = new FileReader(SYMBOL_DESCRIPTION_FILE);
			csvReader = new CSVReader(reader);
			List<String[]> entries = csvReader.readAll();
			Iterator<String[]> i = entries.iterator();

			while (i.hasNext()) {
				String[] tokens = i.next();
				stockNameLoopup.put(tokens[0], tokens[1]);
			}
		}catch (IOException e){
			e.printStackTrace();
		}finally {
			IOUtils.closeQuietly(csvReader);
			IOUtils.closeQuietly(reader);
		}
	}

	@Override
    protected void reduce(DividendsKeyWritable key, Iterable<DividendsValueWritable> values, Context context) throws IOException,
            InterruptedException {
		DividendsValueWritable value =new DividendsValueWritable(); 
		int numOfAnnualDividends = 0;
		for(DividendsValueWritable dividends : values){
			LOG.info("Reducer in action! " + key.toString() + Float.toString(dividends.getAnnualDividendsTotal()));
			value.setAnnualDividendsTotal(value.getAnnualDividendsTotal() + dividends.getAnnualDividendsTotal());
			value.setAnnualCount(value.getAnnualCount() + dividends.getAnnualCount());
			numOfAnnualDividends++;   
		}
		
		LOG.info("Reducer in action end");
		LOG.info("counter " + Integer.toString(numOfAnnualDividends) + " writable " + Integer.toString(value.getAnnualCount()));
		value.setAnnualDividendsTotal(value.getAnnualDividendsTotal() / value.getAnnualCount());
		
		String stockName = null;
		if((stockName = stockNameLoopup.get(key.getSymbol())) != null){
			this.reusableText.set(key.getYearStr()+":" + stockName +"("+key.getSymbol()+")");
			context.getCounter(Tokens.Enriched).increment(1);
		}
		else{
			this.reusableText.set(key.getYearStr()+":("+key.getSymbol()+")");
			context.getCounter(Tokens.NotEnriched).increment(1);
		}
		
		this.reuableFloat.set(value.getAnnualDividendsTotal());
		context.write(this.reusableText, this.reuableFloat);
	}
}
